#include "BigQ.h"
#include <queue>
#include <algorithm>
#include <unistd.h>
#include <sstream>

#ifdef DEBUG
Schema* sch;
#endif

//#define THOONGU 0

PQRecord::PQRecord(int rn,Record* rec){
	runNumber = rn;
	myRec = rec;
}

Run::Run(int totalPg,int off,File* fl){
	//cout << "Run: Total: " << totalPg << " Off: " << off << "\n";
	#ifdef DEBUG
	cout << "Run: Total: " << totalPg << " Off: " << off << "\n";
	#endif
	totalPages = totalPg;
	offset =off;
	file = fl;
	page = NULL;
	currentPage = 0;
	page = new Page();
}

int Run::Load(){
	file->GetPage(page,offset); //offset + currentPage = offset since currentPage = 0
}

int Run::GetNextRecord(Record* rec){
	if(!page->GetFirst(rec)){
		currentPage++;
		if(currentPage < totalPages) {
			file->GetPage(page,(offset + currentPage));
			page->GetFirst(rec);
		} else {
			return 0;
		}
	}
	return 1;
}

void* performSort(void* ptr){
        ((BigQ*)ptr)->doSort();
}

int BigQ::count = 0;
pthread_mutex_t th_mutex = PTHREAD_MUTEX_INITIALIZER;

#ifdef DEBUG
BigQ :: BigQ (Pipe &in, Pipe &out, OrderMaker &sortorder, int runlen,Schema* s) {
	sch = s;
#else
BigQ :: BigQ (Pipe &inP, Pipe &outP, OrderMaker &sortorder, int runlen) {
#endif
if(runlen > 0) {
	// read data from in pipe sort them into runlen pages
	runLength = runlen;
	tmpPages = new Page[runLength];
	fileName= "sort.tmp.";
	stringstream output;
	pthread_mutex_lock(&th_mutex);
	count++;
	output << count;
	pthread_mutex_unlock(&th_mutex);
	fileName = fileName+output.str();
	tmpFile.Open(0,(char*)fileName.c_str());
	currentPage = 0;
	in = &inP;
        out = &outP;
	pageToAdd = 0;
	numberRuns = 0;
	tmpRecordAr = NULL;
	ordMaker = sortorder;
	pthread_create(&thread,NULL,performSort,(void*)this);
}

}


void BigQ::doSort(){
	Record tmpRecord;
	int numRecords = 0;
	Run *tmp;
	while(in->Remove(&tmpRecord)){
		if(!tmpPages[currentPage].Append(&tmpRecord)){
			currentPage++;
			if(currentPage <runLength){
				tmpPages[currentPage].Append(&tmpRecord);
			} else {
				int off = pageToAdd;
				numberRuns++;
				currentPage--;
				sortRunLength(numRecords);
				writeRecordsToFile(numRecords);
				int pgNum = pageToAdd - off;
				tmp = new Run(pgNum,off,&tmpFile);
				runVector.push_back(tmp);
				numRecords = 0;
				tmpPages[currentPage].Append(&tmpRecord);
			}
		}
		numRecords++;
	}

	if(numRecords > 0){
		int off = pageToAdd;	
		numberRuns++;
		sortRunLength(numRecords);
        	writeRecordsToFile(numRecords);
		int pgNum = pageToAdd - off;
		tmp = new Run(pgNum,off,&tmpFile);
		runVector.push_back(tmp);
		numRecords = 0;
	}
	delete[] tmpPages;
    // construct priority queue over sorted runs and dump sorted data 
 	// into the out pipe
	doMerge();
    // finally shut down the out pipe
	out->ShutDown ();
}

void BigQ::sortRunLength(int numRecords){
	//cout << "\n\n------------enter sort length: " << numRecords << " CurrentPage: " << currentPage << "\n\n";
	//sleep(5);
	#ifdef DEBUG
	cout << "\n\n------------enter sort length: " << numRecords << " CurrentPage: " << currentPage << "\n\n";
	#endif
	//tmpRecordAr = new Record*[numRecords];
	int recordNumber = 0;
	Record* tmp;
	for(int pageNum = 0; pageNum<=currentPage; pageNum++){
		tmpPages[pageNum].GoToRecord(0);
		//while(tmpPages[pageNum].GetNextRecordPointer(&tmpRecordAr[recordNumber++]) != 0){
		while(tmpPages[pageNum].GetNextRecordPointer(&tmp) != 0){
			tmpRecordVector.push_back(tmp);
			#ifdef DEBUG
			cout << " recordNumber: " << recordNumber <<  " Val: " << tmpRecordAr[recordNumber-1];
			#endif
		}
		recordNumber--;
	}
	sort(tmpRecordVector.begin(),tmpRecordVector.end(),SortCompare(&ordMaker));
	//qsort(tmpRecordAr,numRecords,sizeof(Record*),doCompare);

	/*for(int i =0;i<numRecords;i++){
		//cout << i << " ";
		out.Insert(tmpRecordAr[i]);
	}
	sleep(10);*/
	#ifdef DEBUG
	cout << "\n\n";
	for(int i =0;i<numRecords;i++){
		cout << " recordNumber: " << i <<  " Val: " << tmpRecordAr[i];
	}
	#endif
}

bool SortCompare::operator()(Record* leftRec, Record* rightRec)const{
	//cout << "Left: " << leftRec << " Right: " << rightRec << "\n";
	//return true;
        int  ret = comeng->Compare(leftRec,rightRec,ord);
	if(ret == -1)
		return true;

	return false;
}

void BigQ::writeRecordsToFile(int numRecords){
	#ifdef DEBUG
	cout << "\nWrite records: " << numRecords << " CurrentPage: " << currentPage  <<  "\n";
	#endif
	#if 1
	int writeToFile = 0;
	for(int i=0;i<numRecords;i++){
		writeToFile = 1;
		#ifdef DEBUG
		cout << "Num: " << i <<"\n";
		#endif

		if(!pageToFile.Append(tmpRecordVector.at(i))){
			#ifdef DEBUG
			cout << "Page added: " << pageToAdd << "\n";
			#endif
			tmpFile.AddPage(&pageToFile,pageToAdd++);
			pageToFile.EmptyItOut();
			pageToFile.Append(tmpRecordVector.at(i));
		}
	}

	if(writeToFile){
		#ifdef DEBUG
		cout << "Page added: " << pageToAdd << "\n";
		#endif
		tmpFile.AddPage(&pageToFile,pageToAdd++);
		pageToFile.EmptyItOut();
	}
	#endif
	//delete[] tmpRecordAr;
	tmpRecordVector.clear();
	delete [] tmpPages;
	tmpPages = new Page[runLength];
	currentPage = 0;
}

void BigQ::doMerge(){
	#ifdef DEBUG
	cout << "\n\n-----------------------------In merge: " << numberRuns << "----------------------------------\n\n";
	#endif
	#ifdef THOONGU
	int a;
	#endif
	int *runAlive = new int[numberRuns];
	PQRecord pqrec;
	MergeCompare *m = new MergeCompare(&ordMaker);
	priority_queue<PQRecord,vector<PQRecord>,MergeCompare> prioQ(*m);

	//Initialize the priority queue
	Record *tmpRec;
	for(int i=0;i<numberRuns;i++){
		tmpRec = new Record();
		runAlive[i] = 1;
		runVector.at(i)->Load();
		runVector.at(i)->GetNextRecord(tmpRec);
		pqrec.runNumber = i;
		pqrec.myRec = tmpRec;
		//Insert the PQRecord into queue
		#ifdef DEBUG
		cout << "Push\n";
		tmpRec->Print(sch);
		#endif
		prioQ.push(pqrec);
	}

	//Pop from queue
	int i=0,a;
	while(!prioQ.empty()){
		pqrec = prioQ.top();
		prioQ.pop();
		//cin >> a;
                out->Insert(pqrec.myRec);
		#ifdef DEBUG
		cout << "In Pop from: " << pqrec.runNumber << "\n";	
		#endif
		#ifdef THOONGU
		if(a != 100)	
		cin >> a;
		else
		 sleep(1);
		#endif
		if(runAlive[pqrec.runNumber]){
			//delete pqrec.myRec;
			tmpRec = new Record();
                       	//if to check for 0, if 0 set runalive[i] to 0,
			if(runVector.at(pqrec.runNumber)->GetNextRecord(tmpRec))
			{
				pqrec.myRec = tmpRec;
				#ifdef DEBUG
				cout <<"Push\n";
				tmpRec->Print(sch);
				#endif
				prioQ.push(pqrec);
			} else {
				runAlive[pqrec.runNumber] = 0;
			}
		}
	}

	for(int i =0;i<numberRuns;i++){
		delete runVector.at(i);
	}
	delete [] runAlive;

}


BigQ::~BigQ () {
	tmpFile.Close();
	remove((char*)fileName.c_str());
}

//lhs < rhs
bool MergeCompare::operator()(const PQRecord& lhs, const PQRecord& rhs) const
{
	//cout << "LHS " << &lhs << " RHS " << &rhs << "\n";
	#ifdef DEBUG
	cout << "In Compare: Left:\n";
	#endif
	#ifdef THOONGU
	sleep(1);
	#endif

	Record* leftRec = lhs.myRec;
	Record* rightRec = rhs.myRec;

	#ifdef DEBUG
	leftRec->Print(sch);
	cout << "Right:\n";
	rightRec->Print(sch);
	#endif
	if(comeng->Compare(leftRec,rightRec,ord) < 0){
	#ifdef DEBUG
		cout << "Less than\n";
	#endif
		return false;
	}
	#ifdef DEBUG
	cout << "Greater than\n";
	#endif
	return true;
}
